package network_programming.np2_nonblocking_communication;

import java.io.*;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;

/**
 * 非阻塞模式 EchoClient     单线程
 * 异步通信
 *
 * 首先通道采用阻塞模式与远程服务器连接，再转为非阻塞模式
 *
 * 使用两个线程，主线程负责接受和发送数据，另一个线程负责读取用户向控制台输入的数据
 * */
public class EchoClientNonblocking {
    private SocketChannel socketChannel;
    private ByteBuffer sendBuffer = ByteBuffer.allocate(1024);
    private ByteBuffer receiveBuffer = ByteBuffer.allocate(1024);
    private Charset charset = StandardCharsets.UTF_8;
    private Selector selector;

    public EchoClientNonblocking() throws IOException {
        socketChannel = SocketChannel.open();
        InetAddress serverAddress = InetAddress.getLocalHost();
        InetSocketAddress socketAddress = new InetSocketAddress(serverAddress, 9099);
        socketChannel.connect(socketAddress);   // 采用阻塞模式连接服务器
        socketChannel.configureBlocking(false); // 设置通道为非阻塞模式
        System.out.println("成功与服务器建立连接");
        selector = Selector.open();
    }

    /**
     * 接收用户从控制台输入的数据，并将数据存放到sendBuffer中
     * */
    public void receiveFromUser() {
        BufferedReader localReader = new BufferedReader(new InputStreamReader(System.in));
        String msg;
        do {
            try {
                if (((msg = localReader.readLine()) != null)) {
                    synchronized (sendBuffer) {
                        sendBuffer.put(charset.encode(msg + "\r\n"));
                    }
                }
                if ("bye".equals(msg))
                    break;
            } catch (IOException e) {
                e.printStackTrace();
            }
        } while (true);
    }

    /**
     * 接收和发送数据
     * */
    public void talk() throws IOException {
        // 注册读就绪、写就绪事件
        socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);

        while (selector.select() > 0) {
            Set<SelectionKey> readyKey = selector.selectedKeys();
            Iterator<SelectionKey> iterator = readyKey.iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                try {
                    if (key.isReadable())
                        receive(key);
                    if (key.isWritable())
                        send(key);
                } catch (IOException e) {
                    e.printStackTrace();
                    if (key != null)
                        try {
                            key.cancel();
                            key.channel().close();
                        } catch (Exception ex) {
                            ex.printStackTrace();
                        }
                }
            }
        }
    }

    private void send(SelectionKey key) throws IOException {
        SocketChannel socketChannel = (SocketChannel) key.channel();
        synchronized(sendBuffer) {
            sendBuffer.flip();  // 把极限设置为位置，把位置设置为0
            socketChannel.write(sendBuffer);
            sendBuffer.compact();   // 删除已经发送的数据
        }
    }

    private void receive(SelectionKey key) throws IOException {
        SocketChannel socketChannel = (SocketChannel) key.channel();
        socketChannel.read(receiveBuffer);
        receiveBuffer.flip();
        String receiveData = charset.decode(receiveBuffer).toString();  // 解码
        if (!receiveData.contains("\n"))
            return;
//        String outputData = receiveData.substring(0, receiveData.indexOf("\n"));
        System.out.print(receiveData);
        if ("echo: bye\r\n".equals(receiveData)) {
            key.channel();
            socketChannel.close();
            System.out.println("关闭与服务器的连接");
            System.exit(0);     // 结束程序
        }
        ByteBuffer temp = charset.encode(receiveData);
        receiveBuffer.position(temp.limit());
        receiveBuffer.compact();
    }

    public static void main(String[] args) throws IOException {
        final EchoClientNonblocking client = new EchoClientNonblocking();
        Thread receiveFormUser = new Thread() {
            @Override
            public void run() {
                client.receiveFromUser();
            }
        };
        receiveFormUser.start();
        client.talk();
    }
}
